package cn.virens.cmpt.redis;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.hutool.core.util.StrUtil;
import cn.virens.app.AppConfig;
import cn.virens.app.cmpt.Component;
import cn.virens.common.utils.common.CloseableUtil;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.api.reactive.RedisReactiveCommands;
import io.lettuce.core.api.sync.RedisCommands;

public class RedisManagerImpl implements RedisManager, Component {
	private final ThreadLocal<StatefulRedisConnection<String, String>> local = new ThreadLocal<>();
	private final Logger logger = LoggerFactory.getLogger(RedisManagerImpl.class);

	private RedisClient client;

	@Override
	public void initialize(AppConfig config) throws Exception {
		RedisURI.Builder builder = RedisURI.builder();
		builder.withHost(config.getString("redis.host"));
		builder.withPort(config.getInteger("redis.port"));
		builder.withDatabase(config.getInteger("redis.database"));

		// 添加用户密码
		String password = config.getString("redis.password");
		if (StrUtil.isNotEmpty(password)) {
			builder.withPassword(password);
		}

		// 打开Redis连接
		this.client = RedisClient.create(builder.build());
	}

	@Override
	public void destroy() throws Exception {
		this.client.shutdown();
	}

	@Override
	public StatefulRedisConnection<String, String> open() {
		StatefulRedisConnection<String, String> conn = local.get();

		// 连接不存在 | 连接被关闭
		if (conn == null || !conn.isOpen()) {
			local.set(conn = client.connect());
		}

		return conn;
	}

	@Override
	public void sync(RedisExecute<RedisCommands<String, String>> execute) {
		RedisCommands<String, String> commands = open().sync();

		try {
			commands.multi();
			execute.exec(commands);
			commands.exec();
		} catch (Exception e) {
			commands.discard();

			// 打印异常信息
			logger.error("Redis 执行异常：" + e.getMessage(), e);
		}
	}

	@Override
	public void async(RedisExecute<RedisAsyncCommands<String, String>> execute) {
		RedisAsyncCommands<String, String> asyncCommands = open().async();

		try {
			asyncCommands.multi();
			execute.exec(asyncCommands);
			asyncCommands.exec();
		} catch (Exception e) {
			asyncCommands.discard();

			// 打印异常信息
			logger.error("Redis 执行异常：" + e.getMessage(), e);
		}
	}

	@Override
	public void reactive(RedisExecute<RedisReactiveCommands<String, String>> execute) {
		RedisReactiveCommands<String, String> reactiveCommands = open().reactive();

		try {
			reactiveCommands.multi();
			execute.exec(reactiveCommands);
			reactiveCommands.exec();
		} catch (Exception e) {
			reactiveCommands.discard();

			// 打印异常信息
			logger.error("Redis 执行异常：" + e.getMessage(), e);
		}
	}

	@Override
	public void close() {
		StatefulRedisConnection<String, String> conn = local.get();
		if (conn != null && conn.isOpen()) {
			CloseableUtil.close(conn);
		}
	}
}
